package com.codejiwei.flink.source;

import com.codejiwei.flink.entity.WaterSensor;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

/**
 * @author jiwei
 * @description
 * @date 2023/5/19 16:16
 */
public class WaterSensorSocketSource implements SourceFunction<WaterSensor> {

    private String host;
    private Integer port;
    private volatile boolean isRunning = true;
    private Socket socket;


    public WaterSensorSocketSource(String host, Integer port) {
        this.host = host;
        this.port = port;
    }

    @Override
    public void run(SourceContext<WaterSensor> ctx) throws Exception {
        socket = new Socket(host, port);
        BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
        String line = null;
        while (isRunning && (line = reader.readLine()) != null) {
            String[] split = line.split(",");
            ctx.collect(new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2])));
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
        try {
            socket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
